package com.sky.disruptor.main;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.sky.disruptor.entity.OrderEntity;
import com.sky.disruptor.exception.OrderHandlerException;
import com.sky.disruptor.factory.OrderFactory;
import com.sky.disruptor.handler.OrderHandler;
import com.sky.disruptor.handler.OrderHandler2;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class OrderMain {

    public static void main(String[] args) throws Exception{

        //danyi();
        for (int i = 0; i < 2; i++) {
            duoge();
        }

    }

    private static void danyi() throws Exception{
        Long time = System.currentTimeMillis();
        OrderFactory orderFactory = new OrderFactory();

        int ringBuffer_size = 1024 * 1024;

        Disruptor<OrderEntity> disruptor = new Disruptor<OrderEntity>(orderFactory, ringBuffer_size, Executors.defaultThreadFactory());

        disruptor.handleEventsWith(new OrderHandler());

        disruptor.start();


        RingBuffer<OrderEntity> ringBuffer = disruptor.getRingBuffer();


        for (int i = 0; i < 100000; i++) {
            long sequence = ringBuffer.next();
            OrderEntity orderEntity = ringBuffer.get(sequence);
            orderEntity.setOrderId("OrderID  = " + i);
            orderEntity.setOrderName("OrderName = " + i);
            ringBuffer.publish(sequence);
            //System.out.println(Thread.currentThread().getName() + "生产一条数据，ID" + i);
        }
        //Thread.sleep(1000);
        disruptor.shutdown();
        System.out.println("单一处理花费时间 :" + (System.currentTimeMillis() - time));
    }

    private static void duoge() throws InterruptedException {
        Long time1 = System.currentTimeMillis();

        // 创建缓冲池
        ExecutorService executor = Executors.newFixedThreadPool(10);
        // 创建工厂
        OrderFactory orderFactory = new OrderFactory();

        int ringBufferSize = 1024 * 1024;

        RingBuffer<OrderEntity> ringBuffer = RingBuffer.create(ProducerType.MULTI, orderFactory, ringBufferSize, new YieldingWaitStrategy());

        SequenceBarrier barriers = ringBuffer.newBarrier();

        // 创建20个消费者
        OrderHandler2[] orderHandler2s = new OrderHandler2[20];
        for (int i = 0; i < orderHandler2s.length; i++) {
            orderHandler2s[i] = new OrderHandler2();
        }

        WorkerPool<OrderEntity> workerPool = new WorkerPool<>(ringBuffer, barriers, new OrderHandlerException(), orderHandler2s);

        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
        workerPool.start(executor);

        for (int i = 0; i < 100000; i++) {
            long sequence = ringBuffer.next();
            OrderEntity orderEntity = ringBuffer.get(sequence);
            orderEntity.setOrderId("OrderID  = " + i);
            orderEntity.setOrderName("OrderName = " + i);
            ringBuffer.publish(sequence);
            //System.out.println(Thread.currentThread().getName() + "生产一条数据，ID" + i);
        }
        //Thread.sleep(1000);e
        workerPool.halt();
        executor.shutdown();
        boolean result = executor.awaitTermination(2000, TimeUnit.MILLISECONDS);
        if (!result) {
            executor.shutdownNow();
        }
        System.out.println("多线程加多个消费者处理花费时间 :" + (System.currentTimeMillis() - time1));
    }
}
